//! 页面管理模块
//!
//! 实现4KB页面式存储管理，包括页面分配、缓存、持久化等功能
//! 这是存储引擎的核心组件，负责管理所有数据页面

use crate::{Error, Result};
use crate::storage::persistence::PersistenceManager;
use std::collections::{HashMap, HashSet};
use std::sync::{Arc, RwLock};
use parking_lot::Mutex;
use serde::{Serialize, Deserialize};
use std::time::SystemTime;
use std::path::{Path, PathBuf};
use std::fs::{File, OpenOptions, create_dir_all};
use std::io::{Read, Write, Seek, SeekFrom};

/// 页面大小常量 - 4KB
pub const PAGE_SIZE: usize = 4096;

/// 页面ID类型
pub type PageId = u64;

/// 页面管理器
/// 负责页面的分配、回收、缓存和持久化
#[derive(Debug)]
pub struct PageManager {
    /// 下一个可用的页面ID
    next_page_id: Arc<Mutex<PageId>>,
    /// 页面缓存 - LRU缓存
    page_cache: Arc<RwLock<LruPageCache>>,
    /// 空闲页面列表
    free_pages: Arc<Mutex<Vec<PageId>>>,
    /// 脏页面集合 - 需要写回磁盘的页面
    dirty_pages: Arc<Mutex<HashSet<PageId>>>,
    /// 页面使用统计
    page_stats: Arc<Mutex<PageStatistics>>,
    /// 持久化管理器
    persistence_manager: Option<Arc<PersistenceManager>>,
    /// 数据文件路径
    data_file_path: Option<PathBuf>,
    /// 数据文件句柄
    data_file: Arc<Mutex<Option<File>>>,
}

/// LRU页面缓存
#[derive(Debug)]
pub struct LruPageCache {
    /// 缓存的页面
    pages: HashMap<PageId, Arc<Page>>,
    /// LRU链表 - 最近使用的页面在前面
    lru_list: Vec<PageId>,
    /// 最大缓存页面数
    max_pages: usize,
}

/// 页面结构 - 4KB固定大小
#[derive(Debug, Clone)]
pub struct Page {
    /// 页面头部信息
    pub header: PageHeader,
    /// 页面数据区域
    pub data: Vec<u8>,
}

/// 页面头部信息 - 24字节
#[repr(C)]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PageHeader {
    /// 页面ID
    pub page_id: PageId,
    /// 页面类型
    pub page_type: PageType,
    /// 日志序列号 - 用于WAL恢复
    pub lsn: u64,
    /// 页面校验和 - 用于数据完整性检查
    pub checksum: u32,
    /// 空闲空间大小
    pub free_space: u16,
    /// 槽数量 - 页面中的记录数
    pub slot_count: u16,
    /// 最后修改时间
    pub last_modified: u64,
}

/// 页面类型
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum PageType {
    /// 数据页 - 存储表数据
    Data,
    /// 索引页 - 存储索引数据
    Index,
    /// 元数据页 - 存储表结构等元数据
    Metadata,
    /// 空闲页 - 未使用的页面
    Free,
    /// WAL页 - 存储事务日志
    Wal,
}

/// 页面统计信息
#[derive(Debug, Default, Clone)]
pub struct PageStatistics {
    /// 总页面数
    pub total_pages: u64,
    /// 已使用页面数
    pub used_pages: u64,
    /// 空闲页面数
    pub free_pages: u64,
    /// 缓存命中次数
    pub cache_hits: u64,
    /// 缓存未命中次数
    pub cache_misses: u64,
    /// 页面读取次数
    pub page_reads: u64,
    /// 页面写入次数
    pub page_writes: u64,
}

impl PageManager {
    /// 创建新的页面管理器
    pub fn new(cache_size: usize) -> Self {
        Self {
            next_page_id: Arc::new(Mutex::new(1)), // 页面ID从1开始
            page_cache: Arc::new(RwLock::new(LruPageCache::new(cache_size))),
            free_pages: Arc::new(Mutex::new(Vec::new())),
            dirty_pages: Arc::new(Mutex::new(HashSet::new())),
            page_stats: Arc::new(Mutex::new(PageStatistics::default())),
            persistence_manager: None,
            data_file_path: None,
            data_file: Arc::new(Mutex::new(None)),
        }
    }

    /// 创建带持久化的页面管理器
    pub fn with_persistence(cache_size: usize, persistence_manager: Arc<PersistenceManager>) -> Self {
        Self {
            next_page_id: Arc::new(Mutex::new(1)),
            page_cache: Arc::new(RwLock::new(LruPageCache::new(cache_size))),
            free_pages: Arc::new(Mutex::new(Vec::new())),
            dirty_pages: Arc::new(Mutex::new(HashSet::new())),
            page_stats: Arc::new(Mutex::new(PageStatistics::default())),
            persistence_manager: Some(persistence_manager),
            data_file_path: None,
            data_file: Arc::new(Mutex::new(None)),
        }
    }

    /// 创建带磁盘存储的页面管理器
    pub fn with_disk_storage<P: AsRef<Path>>(cache_size: usize, data_file_path: P) -> Result<Self> {
        let path = data_file_path.as_ref().to_path_buf();

        // 确保目录存在
        if let Some(parent) = path.parent() {
            create_dir_all(parent)
                .map_err(|e| Error::storage(format!("创建数据目录失败: {}", e)))?;
        }

        // 打开或创建数据文件
        let file = OpenOptions::new()
            .read(true)
            .write(true)
            .create(true)
            .open(&path)
            .map_err(|e| Error::storage(format!("打开数据文件失败: {}", e)))?;

        Ok(Self {
            next_page_id: Arc::new(Mutex::new(1)),
            page_cache: Arc::new(RwLock::new(LruPageCache::new(cache_size))),
            free_pages: Arc::new(Mutex::new(Vec::new())),
            dirty_pages: Arc::new(Mutex::new(HashSet::new())),
            page_stats: Arc::new(Mutex::new(PageStatistics::default())),
            persistence_manager: None,
            data_file_path: Some(path),
            data_file: Arc::new(Mutex::new(Some(file))),
        })
    }
    
    /// 分配新页面
    /// 优先使用空闲页面，如果没有则创建新页面
    pub fn allocate_page(&self, page_type: PageType) -> Result<PageId> {
        // 首先尝试从空闲页面列表获取
        let page_id = {
            let mut free_pages = self.free_pages.lock();
            if let Some(page_id) = free_pages.pop() {
                page_id
            } else {
                // 没有空闲页面，分配新的页面ID
                let mut next_id = self.next_page_id.lock();
                let page_id = *next_id;
                *next_id += 1;
                page_id
            }
        };
        
        // 创建新页面
        let page = Page::new(page_id, page_type);
        
        // 添加到缓存
        {
            let mut cache = self.page_cache.write().unwrap();
            cache.insert(page_id, Arc::new(page));
        }
        
        // 更新统计信息
        {
            let mut stats = self.page_stats.lock();
            stats.total_pages += 1;
            stats.used_pages += 1;
        }
        
        tracing::debug!("分配新页面: {} (类型: {:?})", page_id, page_type);
        Ok(page_id)
    }
    
    /// 释放页面
    /// 将页面标记为空闲，可以被重新分配
    pub fn deallocate_page(&self, page_id: PageId) -> Result<()> {
        // 从缓存中移除
        {
            let mut cache = self.page_cache.write().unwrap();
            cache.remove(page_id);
        }
        
        // 从脏页面集合中移除
        {
            let mut dirty_pages = self.dirty_pages.lock();
            dirty_pages.remove(&page_id);
        }
        
        // 添加到空闲页面列表
        {
            let mut free_pages = self.free_pages.lock();
            free_pages.push(page_id);
        }
        
        // 更新统计信息
        {
            let mut stats = self.page_stats.lock();
            stats.used_pages = stats.used_pages.saturating_sub(1);
            stats.free_pages += 1;
        }
        
        tracing::debug!("释放页面: {}", page_id);
        Ok(())
    }
    
    /// 读取页面
    /// 首先从缓存查找，如果不存在则从磁盘加载
    pub fn read_page(&self, page_id: PageId) -> Result<Arc<Page>> {
        // 首先尝试从缓存获取
        {
            let mut cache = self.page_cache.write().unwrap();
            if let Some(page) = cache.get(page_id) {
                // 缓存命中
                let mut stats = self.page_stats.lock();
                stats.cache_hits += 1;
                return Ok(page);
            }
        }
        
        // 缓存未命中，需要从磁盘加载
        let page = self.load_page_from_disk(page_id)?;
        
        // 添加到缓存
        {
            let mut cache = self.page_cache.write().unwrap();
            cache.insert(page_id, page.clone());
        }
        
        // 更新统计信息
        {
            let mut stats = self.page_stats.lock();
            stats.cache_misses += 1;
            stats.page_reads += 1;
        }
        
        Ok(page)
    }
    
    /// 写入页面
    /// 将页面标记为脏页面，稍后批量写回磁盘
    pub fn write_page(&self, page: Arc<Page>) -> Result<()> {
        let page_id = page.header.page_id;
        
        // 更新缓存中的页面
        {
            let mut cache = self.page_cache.write().unwrap();
            cache.insert(page_id, page);
        }
        
        // 标记为脏页面
        {
            let mut dirty_pages = self.dirty_pages.lock();
            dirty_pages.insert(page_id);
        }
        
        tracing::debug!("页面标记为脏页面: {}", page_id);
        Ok(())
    }
    
    /// 刷新脏页面到磁盘
    /// 将所有脏页面写回磁盘
    pub fn flush_dirty_pages(&self) -> Result<()> {
        let dirty_page_ids: Vec<PageId> = {
            let dirty_pages = self.dirty_pages.lock();
            dirty_pages.iter().cloned().collect()
        };
        
        let dirty_count = dirty_page_ids.len();

        for page_id in &dirty_page_ids {
            let page = {
                let cache = self.page_cache.read().unwrap();
                cache.pages.get(&page_id).cloned()
            };
            
            if let Some(page) = page {
                self.write_page_to_disk(&page)?;

                // 从脏页面集合中移除
                let mut dirty_pages = self.dirty_pages.lock();
                dirty_pages.remove(page_id);
                
                // 更新统计信息
                let mut stats = self.page_stats.lock();
                stats.page_writes += 1;
            }
        }
        
        tracing::debug!("刷新了 {} 个脏页面", dirty_count);
        Ok(())
    }
    
    /// 从磁盘加载页面
    /// 实现真正的磁盘I/O读取功能
    fn load_page_from_disk(&self, page_id: PageId) -> Result<Arc<Page>> {
        tracing::debug!("从磁盘加载页面: {}", page_id);

        // 如果没有数据文件，创建空页面
        if self.data_file_path.is_none() {
            let page = Page::new(page_id, PageType::Data);
            return Ok(Arc::new(page));
        }

        // 计算页面在文件中的偏移量
        let offset = page_id * PAGE_SIZE as u64;

        // 读取页面数据
        let mut data_file = self.data_file.lock();
        if let Some(ref mut file) = *data_file {
            // 定位到页面位置
            file.seek(SeekFrom::Start(offset))
                .map_err(|e| Error::storage(format!("定位页面{}失败: {}", page_id, e)))?;

            // 读取页面数据
            let mut buffer = vec![0u8; PAGE_SIZE];
            match file.read_exact(&mut buffer) {
                Ok(_) => {
                    // 反序列化页面
                    match self.deserialize_page(&buffer) {
                        Ok(page) => Ok(Arc::new(page)),
                        Err(_) => {
                            // 反序列化失败，可能是新页面，创建空页面
                            tracing::warn!("页面{}反序列化失败，创建新页面", page_id);
                            let page = Page::new(page_id, PageType::Data);
                            Ok(Arc::new(page))
                        }
                    }
                }
                Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
                    // 文件末尾，创建新页面
                    tracing::debug!("页面{}超出文件末尾，创建新页面", page_id);
                    let page = Page::new(page_id, PageType::Data);
                    Ok(Arc::new(page))
                }
                Err(e) => Err(Error::storage(format!("读取页面{}失败: {}", page_id, e)))
            }
        } else {
            // 文件未打开，创建空页面
            let page = Page::new(page_id, PageType::Data);
            Ok(Arc::new(page))
        }
    }
    
    /// 将页面写入磁盘
    /// 实现真正的磁盘I/O写入功能
    fn write_page_to_disk(&self, page: &Page) -> Result<()> {
        tracing::debug!("将页面写入磁盘: {}", page.header.page_id);

        // 如果没有数据文件路径，跳过写入
        if self.data_file_path.is_none() {
            return Ok(());
        }

        // 序列化页面数据
        let data = self.serialize_page(page)?;

        // 计算页面在文件中的偏移量
        let offset = page.header.page_id * PAGE_SIZE as u64;

        // 写入页面数据
        let mut data_file = self.data_file.lock();
        if let Some(ref mut file) = *data_file {
            // 定位到页面位置
            file.seek(SeekFrom::Start(offset))
                .map_err(|e| Error::storage(format!("定位页面{}失败: {}", page.header.page_id, e)))?;

            // 写入页面数据
            file.write_all(&data)
                .map_err(|e| Error::storage(format!("写入页面{}失败: {}", page.header.page_id, e)))?;

            // 强制刷新到磁盘
            file.sync_all()
                .map_err(|e| Error::storage(format!("同步页面{}失败: {}", page.header.page_id, e)))?;

            tracing::debug!("页面{}成功写入磁盘", page.header.page_id);
        }

        Ok(())
    }

    /// 序列化页面到字节数组
    /// 将页面结构转换为可存储的二进制格式
    fn serialize_page(&self, page: &Page) -> Result<Vec<u8>> {
        let mut buffer = vec![0u8; PAGE_SIZE];

        // 手动序列化页面头部 (固定32字节)
        let mut offset = 0;

        // 页面ID (8字节)
        buffer[offset..offset + 8].copy_from_slice(&page.header.page_id.to_le_bytes());
        offset += 8;

        // 页面类型 (1字节)
        buffer[offset] = match page.header.page_type {
            PageType::Data => 0,
            PageType::Index => 1,
            PageType::Metadata => 2,
            PageType::Free => 3,
            PageType::Wal => 4,
        };
        offset += 1;

        // LSN (8字节)
        buffer[offset..offset + 8].copy_from_slice(&page.header.lsn.to_le_bytes());
        offset += 8;

        // 校验和 (4字节)
        buffer[offset..offset + 4].copy_from_slice(&page.header.checksum.to_le_bytes());
        offset += 4;

        // 空闲空间 (2字节)
        buffer[offset..offset + 2].copy_from_slice(&page.header.free_space.to_le_bytes());
        offset += 2;

        // 槽数量 (2字节)
        buffer[offset..offset + 2].copy_from_slice(&page.header.slot_count.to_le_bytes());
        offset += 2;

        // 最后修改时间 (8字节) - 但我们只有1字节剩余，所以跳过
        // offset现在是31，还有1字节到32
        // 填充剩余字节
        while offset < 32 {
            buffer[offset] = 0;
            offset += 1;
        }

        // 复制页面数据
        let data_start = 32;
        let data_len = std::cmp::min(page.data.len(), PAGE_SIZE - data_start);
        buffer[data_start..data_start + data_len].copy_from_slice(&page.data[..data_len]);

        Ok(buffer)
    }

    /// 从字节数组反序列化页面
    /// 将二进制数据转换为页面结构
    fn deserialize_page(&self, buffer: &[u8]) -> Result<Page> {
        if buffer.len() != PAGE_SIZE {
            return Err(Error::storage(format!("页面数据大小错误: {} != {}", buffer.len(), PAGE_SIZE)));
        }

        // 手动反序列化页面头部
        let mut offset = 0;

        // 页面ID (8字节)
        let page_id = u64::from_le_bytes([
            buffer[offset], buffer[offset + 1], buffer[offset + 2], buffer[offset + 3],
            buffer[offset + 4], buffer[offset + 5], buffer[offset + 6], buffer[offset + 7],
        ]);
        offset += 8;

        // 页面类型 (1字节)
        let page_type = match buffer[offset] {
            0 => PageType::Data,
            1 => PageType::Index,
            2 => PageType::Metadata,
            3 => PageType::Free,
            4 => PageType::Wal,
            _ => PageType::Data, // 默认为数据页
        };
        offset += 1;

        // LSN (8字节)
        let lsn = u64::from_le_bytes([
            buffer[offset], buffer[offset + 1], buffer[offset + 2], buffer[offset + 3],
            buffer[offset + 4], buffer[offset + 5], buffer[offset + 6], buffer[offset + 7],
        ]);
        offset += 8;

        // 校验和 (4字节)
        let checksum = u32::from_le_bytes([
            buffer[offset], buffer[offset + 1], buffer[offset + 2], buffer[offset + 3],
        ]);
        offset += 4;

        // 空闲空间 (2字节)
        let free_space = u16::from_le_bytes([buffer[offset], buffer[offset + 1]]);
        offset += 2;

        // 槽数量 (2字节)
        let slot_count = u16::from_le_bytes([buffer[offset], buffer[offset + 1]]);
        offset += 2;

        // 最后修改时间 - 使用当前时间
        let last_modified = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_secs();

        let header = PageHeader {
            page_id,
            page_type,
            lsn,
            checksum,
            free_space,
            slot_count,
            last_modified,
        };

        // 提取页面数据
        let data = buffer[32..].to_vec();

        Ok(Page {
            header,
            data,
        })
    }

    /// 获取页面统计信息
    pub fn get_statistics(&self) -> PageStatistics {
        (*self.page_stats.lock()).clone()
    }

    /// 获取缓存命中率
    pub fn get_cache_hit_rate(&self) -> f64 {
        let stats = self.page_stats.lock();
        let total_accesses = stats.cache_hits + stats.cache_misses;
        if total_accesses > 0 {
            stats.cache_hits as f64 / total_accesses as f64
        } else {
            0.0
        }
    }

    /// 获取页面（支持测试用例）
    pub fn get_page(&self, page_id: PageId) -> Result<Arc<Page>> {
        self.read_page(page_id)
    }

    /// 获取可变页面引用（支持测试用例）
    pub fn get_page_mut(&self, page_id: PageId) -> Result<Arc<Page>> {
        // 注意：在实际实现中，这里应该返回一个可以修改的页面引用
        // 为了简化，我们先返回普通的页面引用
        self.read_page(page_id)
    }

    /// 获取缓存大小
    pub fn cache_size(&self) -> usize {
        let cache = self.page_cache.read().unwrap();
        cache.max_pages
    }

    /// 获取当前缓存的页面数量
    pub fn cached_pages_count(&self) -> usize {
        let cache = self.page_cache.read().unwrap();
        cache.size()
    }
}

impl LruPageCache {
    /// 创建新的LRU缓存
    pub fn new(max_pages: usize) -> Self {
        Self {
            pages: HashMap::new(),
            lru_list: Vec::new(),
            max_pages,
        }
    }

    /// 插入页面到缓存
    pub fn insert(&mut self, page_id: PageId, page: Arc<Page>) {
        // 如果页面已存在，更新LRU位置
        if self.pages.contains_key(&page_id) {
            self.move_to_front(page_id);
            self.pages.insert(page_id, page);
            return;
        }

        // 如果缓存已满，移除最久未使用的页面
        if self.pages.len() >= self.max_pages {
            self.evict_lru();
        }

        // 插入新页面
        self.pages.insert(page_id, page);
        self.lru_list.insert(0, page_id);
    }

    /// 获取页面（不更新LRU顺序，用于只读访问）
    pub fn get_readonly(&self, page_id: PageId) -> Option<Arc<Page>> {
        self.pages.get(&page_id).cloned()
    }

    /// 获取页面（更新LRU顺序）
    pub fn get(&mut self, page_id: PageId) -> Option<Arc<Page>> {
        if self.pages.contains_key(&page_id) {
            // 移动到LRU列表前端
            self.move_to_front(page_id);
            // 获取页面
            self.pages.get(&page_id).cloned()
        } else {
            None
        }
    }

    /// 移除页面
    pub fn remove(&mut self, page_id: PageId) -> Option<Arc<Page>> {
        if let Some(page) = self.pages.remove(&page_id) {
            // 从LRU列表中移除
            self.lru_list.retain(|&id| id != page_id);
            Some(page)
        } else {
            None
        }
    }

    /// 将页面移动到LRU列表前端
    fn move_to_front(&mut self, page_id: PageId) {
        // 从当前位置移除
        self.lru_list.retain(|&id| id != page_id);
        // 插入到前端
        self.lru_list.insert(0, page_id);
    }

    /// 淘汰最久未使用的页面
    fn evict_lru(&mut self) {
        if let Some(lru_page_id) = self.lru_list.pop() {
            self.pages.remove(&lru_page_id);
            tracing::debug!("淘汰LRU页面: {}", lru_page_id);
        }
    }

    /// 获取缓存大小
    pub fn size(&self) -> usize {
        self.pages.len()
    }

    /// 清空缓存
    pub fn clear(&mut self) {
        self.pages.clear();
        self.lru_list.clear();
    }
}

impl Page {
    /// 创建新页面
    pub fn new(page_id: PageId, page_type: PageType) -> Self {
        let header = PageHeader {
            page_id,
            page_type,
            lsn: 0,
            checksum: 0,
            free_space: (PAGE_SIZE - std::mem::size_of::<PageHeader>()) as u16,
            slot_count: 0,
            last_modified: SystemTime::now()
                .duration_since(SystemTime::UNIX_EPOCH)
                .unwrap()
                .as_secs(),
        };

        let data = vec![0u8; PAGE_SIZE - std::mem::size_of::<PageHeader>()];

        Self { header, data }
    }

    /// 计算页面校验和
    pub fn calculate_checksum(&self) -> u32 {
        // 简化的校验和算法 - 实际应该使用CRC32
        let mut checksum = 0u32;
        for byte in &self.data {
            checksum = checksum.wrapping_add(*byte as u32);
        }
        checksum
    }

    /// 验证页面校验和
    pub fn verify_checksum(&self) -> bool {
        self.header.checksum == self.calculate_checksum()
    }

    /// 更新页面校验和
    pub fn update_checksum(&mut self) {
        self.header.checksum = self.calculate_checksum();
    }

    /// 获取可用空间大小
    pub fn available_space(&self) -> usize {
        self.header.free_space as usize
    }

    /// 标记页面为已修改
    pub fn mark_dirty(&mut self) {
        self.header.last_modified = SystemTime::now()
            .duration_since(SystemTime::UNIX_EPOCH)
            .unwrap()
            .as_secs();
        self.update_checksum();
    }

    /// 获取页面数据的可变引用
    pub fn data_mut(&mut self) -> &mut [u8] {
        &mut self.data
    }

    /// 获取页面数据的不可变引用
    pub fn data(&self) -> &[u8] {
        &self.data
    }
}

impl Default for PageHeader {
    fn default() -> Self {
        Self {
            page_id: 0,
            page_type: PageType::Free,
            lsn: 0,
            checksum: 0,
            free_space: 0,
            slot_count: 0,
            last_modified: 0,
        }
    }
}

impl PageStatistics {
    /// 获取缓存命中率
    pub fn cache_hit_rate(&self) -> f64 {
        let total = self.cache_hits + self.cache_misses;
        if total > 0 {
            self.cache_hits as f64 / total as f64
        } else {
            0.0
        }
    }

    /// 获取页面使用率
    pub fn page_usage_rate(&self) -> f64 {
        if self.total_pages > 0 {
            self.used_pages as f64 / self.total_pages as f64
        } else {
            0.0
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_page_manager_creation() {
        let page_manager = PageManager::new(100);
        let stats = page_manager.get_statistics();
        assert_eq!(stats.total_pages, 0);
        assert_eq!(stats.used_pages, 0);
    }

    #[test]
    fn test_page_allocation() {
        let page_manager = PageManager::new(100);

        // 分配数据页
        let page_id = page_manager.allocate_page(PageType::Data).unwrap();
        assert_eq!(page_id, 1);

        // 分配索引页
        let page_id2 = page_manager.allocate_page(PageType::Index).unwrap();
        assert_eq!(page_id2, 2);

        let stats = page_manager.get_statistics();
        assert_eq!(stats.total_pages, 2);
        assert_eq!(stats.used_pages, 2);
    }

    #[test]
    fn test_page_deallocation() {
        let page_manager = PageManager::new(100);

        let page_id = page_manager.allocate_page(PageType::Data).unwrap();
        page_manager.deallocate_page(page_id).unwrap();

        let stats = page_manager.get_statistics();
        assert_eq!(stats.used_pages, 0);
        assert_eq!(stats.free_pages, 1);
    }

    #[test]
    fn test_page_read_write() {
        let page_manager = PageManager::new(100);

        let page_id = page_manager.allocate_page(PageType::Data).unwrap();

        // 读取页面
        let page = page_manager.read_page(page_id).unwrap();
        assert_eq!(page.header.page_id, page_id);
        assert_eq!(page.header.page_type, PageType::Data);

        // 写入页面
        page_manager.write_page(page).unwrap();

        let stats = page_manager.get_statistics();
        assert_eq!(stats.cache_hits, 1);
    }

    #[test]
    fn test_lru_cache() {
        let mut cache = LruPageCache::new(2);

        let page1 = Arc::new(Page::new(1, PageType::Data));
        let page2 = Arc::new(Page::new(2, PageType::Data));
        let page3 = Arc::new(Page::new(3, PageType::Data));

        // 插入两个页面
        cache.insert(1, page1.clone());
        cache.insert(2, page2.clone());
        assert_eq!(cache.size(), 2);

        // 访问页面1
        let retrieved = cache.get(1);
        assert!(retrieved.is_some());

        // 插入第三个页面，应该淘汰页面2
        cache.insert(3, page3.clone());
        assert_eq!(cache.size(), 2);

        // 页面2应该被淘汰
        let retrieved = cache.get(2);
        assert!(retrieved.is_none());

        // 页面1和3应该还在
        assert!(cache.get(1).is_some());
        assert!(cache.get(3).is_some());
    }

    #[test]
    fn test_page_checksum() {
        let mut page = Page::new(1, PageType::Data);

        // 修改数据
        page.data_mut()[0] = 42;
        page.update_checksum();

        // 验证校验和
        assert!(page.verify_checksum());

        // 破坏数据
        page.data_mut()[0] = 43;

        // 校验和应该不匹配
        assert!(!page.verify_checksum());
    }

    #[test]
    fn test_page_statistics() {
        let stats = PageStatistics {
            cache_hits: 80,
            cache_misses: 20,
            total_pages: 100,
            used_pages: 75,
            ..Default::default()
        };

        assert_eq!(stats.cache_hit_rate(), 0.8);
        assert_eq!(stats.page_usage_rate(), 0.75);
    }

    #[test]
    fn test_page_manager_advanced_operations() {
        let manager = PageManager::new(10);

        // 测试页面分配和访问
        let page_id = manager.allocate_page(PageType::Data).unwrap();
        let page = manager.get_page(page_id).unwrap();
        assert_eq!(page.header.page_id, page_id);
        assert_eq!(page.header.page_type, PageType::Data);

        // 测试缓存大小
        assert_eq!(manager.cache_size(), 10);
        assert!(manager.cached_pages_count() <= 10);
    }

    #[test]
    fn test_page_manager_dirty_page_management() {
        let manager = PageManager::new(5);

        // 分配页面并修改
        let page_id = manager.allocate_page(PageType::Data).unwrap();
        let page = manager.get_page_mut(page_id).unwrap();

        // 写入页面（标记为脏页）
        manager.write_page(page).unwrap();

        // 刷新脏页
        manager.flush_dirty_pages().unwrap();

        // 验证统计信息
        let stats = manager.get_statistics();
        assert!(stats.page_writes > 0);
    }

    #[test]
    fn test_page_manager_cache_behavior() {
        let manager = PageManager::new(3); // 小缓存测试

        // 分配多个页面
        let page1 = manager.allocate_page(PageType::Data).unwrap();
        let page2 = manager.allocate_page(PageType::Data).unwrap();
        let page3 = manager.allocate_page(PageType::Data).unwrap();
        let page4 = manager.allocate_page(PageType::Data).unwrap();

        // 访问页面，测试缓存行为
        let _ = manager.get_page(page1).unwrap();
        let _ = manager.get_page(page2).unwrap();
        let _ = manager.get_page(page3).unwrap();
        let _ = manager.get_page(page4).unwrap(); // 这应该触发缓存淘汰

        // 验证缓存大小限制
        assert!(manager.cached_pages_count() <= 3);

        // 验证缓存命中率统计
        let hit_rate = manager.get_cache_hit_rate();
        assert!(hit_rate >= 0.0 && hit_rate <= 1.0);
    }
}

// ==================== 持久化相关方法 ====================

impl PageManager {
    /// 异步写入页面到持久化存储
    pub async fn write_page_persistent(&self, page: Arc<Page>) -> Result<()> {
        if let Some(ref persistence) = self.persistence_manager {
            persistence.write_page(page.header.page_id, &page).await?;

            // 从脏页面集合中移除
            {
                let mut dirty_pages = self.dirty_pages.lock();
                dirty_pages.remove(&page.header.page_id);
            }

            tracing::debug!("页面 {} 已持久化", page.header.page_id);
        }
        Ok(())
    }

    /// 异步读取页面（支持持久化）
    pub async fn read_page_persistent(&self, page_id: PageId) -> Result<Option<Arc<Page>>> {
        // 首先检查缓存
        {
            let cache = self.page_cache.read().unwrap();
            if let Some(page) = cache.get_readonly(page_id) {
                // 更新统计信息
                {
                    let mut stats = self.page_stats.lock();
                    stats.cache_hits += 1;
                }

                tracing::debug!("从缓存获取页面: {}", page_id);
                return Ok(Some(page));
            }
        }

        // 更新缓存未命中统计
        {
            let mut stats = self.page_stats.lock();
            stats.cache_misses += 1;
        }

        // 页面不在缓存中，尝试从持久化存储加载
        if let Some(ref persistence) = self.persistence_manager {
            if let Some(page) = persistence.read_page(page_id).await? {
                let page_arc = Arc::new(page);

                // 添加到缓存
                {
                    let mut cache = self.page_cache.write().unwrap();
                    cache.insert(page_id, page_arc.clone());
                }

                tracing::debug!("从持久化存储加载页面: {}", page_id);
                return Ok(Some(page_arc));
            }
        }

        tracing::debug!("页面 {} 不存在", page_id);
        Ok(None)
    }



    /// 设置持久化管理器
    pub fn set_persistence_manager(&mut self, persistence_manager: Arc<PersistenceManager>) {
        self.persistence_manager = Some(persistence_manager);
    }

    /// 检查是否启用持久化
    pub fn has_persistence(&self) -> bool {
        self.persistence_manager.is_some()
    }
}

// 包含磁盘持久化测试
#[cfg(test)]
mod disk_persistence_tests {
    use super::*;
    use tempfile::TempDir;

    /// 测试磁盘存储功能
    #[test]
    fn test_disk_storage() -> Result<()> {
        let temp_dir = TempDir::new()
            .map_err(|e| Error::storage(format!("创建临时目录失败: {}", e)))?;
        let data_file = temp_dir.path().join("test.db");

        // 创建页面管理器并分配页面
        let page_manager = PageManager::with_disk_storage(10, &data_file)?;

        let page_id1 = page_manager.allocate_page(PageType::Data)?;
        let page_id2 = page_manager.allocate_page(PageType::Index)?;

        // 读取页面并修改数据
        let page1 = page_manager.read_page(page_id1)?;
        let page2 = page_manager.read_page(page_id2)?;

        // 创建可修改的页面副本
        let mut page1_copy = (*page1).clone();
        let mut page2_copy = (*page2).clone();

        // 修改页面数据
        page1_copy.data[0] = 0xAA;
        page1_copy.data[1] = 0xBB;
        page2_copy.data[0] = 0xCC;
        page2_copy.data[1] = 0xDD;

        // 写入页面
        page_manager.write_page(Arc::new(page1_copy))?;
        page_manager.write_page(Arc::new(page2_copy))?;

        // 刷新到磁盘
        page_manager.flush_dirty_pages()?;

        // 验证文件存在且有内容
        assert!(data_file.exists());
        let file_size = std::fs::metadata(&data_file)
            .map_err(|e| Error::storage(format!("获取文件大小失败: {}", e)))?
            .len();
        assert!(file_size > 0);

        Ok(())
    }

    /// 测试数据恢复功能
    #[test]
    fn test_data_recovery() -> Result<()> {
        let temp_dir = TempDir::new()
            .map_err(|e| Error::storage(format!("创建临时目录失败: {}", e)))?;
        let data_file = temp_dir.path().join("test.db");

        // 第一阶段：创建并写入数据
        {
            let page_manager = PageManager::with_disk_storage(10, &data_file)?;

            let page_id1 = page_manager.allocate_page(PageType::Data)?;
            let page_id2 = page_manager.allocate_page(PageType::Index)?;

            let page1 = page_manager.read_page(page_id1)?;
            let page2 = page_manager.read_page(page_id2)?;

            // 创建可修改的页面副本
            let mut page1_copy = (*page1).clone();
            let mut page2_copy = (*page2).clone();

            // 写入特定数据
            page1_copy.data[0] = 0x11;
            page1_copy.data[1] = 0x22;
            page2_copy.data[0] = 0x33;
            page2_copy.data[1] = 0x44;

            page_manager.write_page(Arc::new(page1_copy))?;
            page_manager.write_page(Arc::new(page2_copy))?;
            page_manager.flush_dirty_pages()?;
        }

        // 第二阶段：重新创建页面管理器并验证数据恢复
        {
            let page_manager = PageManager::with_disk_storage(10, &data_file)?;

            let page1 = page_manager.read_page(1)?;
            let page2 = page_manager.read_page(2)?;

            // 验证页面类型
            assert_eq!(page1.header.page_type, PageType::Data);
            assert_eq!(page2.header.page_type, PageType::Index);

            // 验证数据内容
            assert_eq!(page1.data[0], 0x11);
            assert_eq!(page1.data[1], 0x22);
            assert_eq!(page2.data[0], 0x33);
            assert_eq!(page2.data[1], 0x44);
        }

        Ok(())
    }

    /// 测试页面序列化和反序列化
    #[test]
    fn test_page_serialization() -> Result<()> {
        let temp_dir = TempDir::new()
            .map_err(|e| Error::storage(format!("创建临时目录失败: {}", e)))?;
        let data_file = temp_dir.path().join("test.db");

        let page_manager = PageManager::with_disk_storage(10, &data_file)?;

        // 创建页面并设置特定数据
        let page_id = page_manager.allocate_page(PageType::Data)?;
        let page = page_manager.read_page(page_id)?;

        // 创建可修改的页面副本
        let mut page_copy = (*page).clone();

        // 设置测试数据
        for i in 0..100 {
            page_copy.data[i] = (i % 256) as u8;
        }

        // 写入并刷新
        page_manager.write_page(Arc::new(page_copy))?;
        page_manager.flush_dirty_pages()?;

        // 重新创建页面管理器并读取
        let page_manager2 = PageManager::with_disk_storage(10, &data_file)?;
        let recovered_page = page_manager2.read_page(page_id)?;

        // 验证数据完整性
        for i in 0..100 {
            assert_eq!(recovered_page.data[i], (i % 256) as u8);
        }

        Ok(())
    }
}
